package com.zoo.curator;

import com.jz.myenum.ECuratorType;
import com.jz.util.StringUtil;
import com.zoo.curator.impl.LocalLeaderSelectorClient;
import com.zoo.curator.impl.MasterCurator;
import com.zoo.util.CuratorSupport;
import com.jz.util.AttributeSupport;
import com.jz.util.JAXBUtils;
import com.jz.util.TimeUtil;
import com.jz.bean.AppInfoBean;
import com.jz.bean.SynchronizedAppConfigBean;
import com.jz.bean.SynchronizedMainConfigBean;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.api.CuratorEvent;
import org.apache.curator.framework.api.CuratorListener;
import org.apache.curator.framework.recipes.cache.*;
import org.apache.curator.framework.state.ConnectionState;
import org.apache.curator.framework.state.ConnectionStateListener;
import org.apache.zookeeper.CreateMode;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.*;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * Created by Administrator on 2015/10/29.
 */
abstract public class AbstractCurator implements Runnable {
    public static final Logger LOGGER = LoggerFactory.getLogger(AbstractCurator.class);
    protected CuratorFramework client = null;
    protected String exec[] = null;
    protected Process child = null;
    protected AbstractLeaderSelectorClient leaderSelector = null;
    protected boolean master = false;

    public AbstractCurator(String execStr, boolean master) {
        this.exec = StringUtil.convertString2Array(execStr,"###");
        this.master = master;
        init();
    }

    public void close() {
        if (client != null) {
            client.close();
        }
    }

    private void init() {
        try {
            client = CuratorSupport.newClient();
            client.start();
            if (master) {
                leaderSelector = new LocalLeaderSelectorClient(this, AttributeSupport.APP_NODE);
            }
            initData();
            initListener();
            client.getCuratorListenable().addListener(getCuratorListener());
            client.getConnectionStateListenable().addListener(getConnectionStateListener());
        } catch (Exception e) {
            LOGGER.error("初始化异常", e);
        }
    }

    private void initData() {
        try {
            if (client == null) {
                return;
            }
            //建立主节点
//            AttributeSupport.deleteNode(client, AttributeSupport.MAIN_NODE);
            String mainXML = CuratorSupport.getMainNodeInitValue();
            CuratorSupport.createNode(client, AttributeSupport.MAIN_NODE, CreateMode.PERSISTENT, mainXML.getBytes());
            synchronizedMainConfig(mainXML);

            //建立APP节点
            String appXML = CuratorSupport.getAppNodeInitValue();
            CuratorSupport.createNode(client, AttributeSupport.APP_NODE, CreateMode.PERSISTENT, appXML.getBytes());
            updateAppNodeValue(new AppInfoBean(false, TimeUtil.getTime(), "zk service start and been watched"));
            synchronizedAppConfig(new String(CuratorSupport.getNodeValue(client, AttributeSupport.APP_NODE)));
            refreshAPP_NODEStatus();
        } catch (Exception e) {
            LOGGER.error("建立节点异常", e);
            System.exit(1);
        }
    }

    /**
     * 定时刷新本机服务的APP_NODE状态
     */
    private void refreshAPP_NODEStatus() {
        //定时更新本机APP_NODE状态
        new Thread(new Runnable() {
            public void run() {
                try {
                    Thread.sleep(1000 * 5);
                    updateAppNodeValue(new AppInfoBean(child != null, TimeUtil.getTime(), "local app node update"));
                } catch (Exception e) {
                    LOGGER.error("", e);
                }
            }
        }).start();
    }

    /**
     * 更新APP_NODE节点的数据
     */
    public void updateAppNodeValue(AppInfoBean bean) {
        bean.setCuratorType(getCuratorType());
        CuratorSupport.localAppInfo = bean;
        SynchronizedAppConfigBean appConfig = null;
        byte[] data = CuratorSupport.getNodeValue(client, AttributeSupport.APP_NODE);
        try {
            if (data == null) {
                appConfig = new SynchronizedAppConfigBean();
            } else {
                String temp = new String(data, "UTF-8");
                appConfig = JAXBUtils.xmlToDto(temp, SynchronizedAppConfigBean.class);
            }
        } catch (Exception e) {
            LOGGER.error("读取APP节点数据格式异常,重置数据", e);
            appConfig = new SynchronizedAppConfigBean();
        }

        /**
         * 定时刷新,判断非本机的APP_NODE节点状态是否正常
         * */
        try {
            boolean found = false;
            for (AppInfoBean i : appConfig.getList()) {
                if (bean.getLocalIP().trim().equals(i.getLocalIP().trim())) {
                    found = true;
                    i.setMessage(bean.getMessage());
                    i.setChildRun(bean.isChildRun());
                    i.setUpdateTime(bean.getUpdateTime());
                } else {
                    //将其他APP节点的更新时间早于当前时间60*2秒以上的节点状态进行更新
                    if (TimeUtil.beforeNow(i.getUpdateTime(), 120)) {
                        i.setMessage(bean.getLocalIP() + ":No update status within a minute");
                        i.setUpdateTime(TimeUtil.getTime());
                        i.setChildRun(false);
                    }
                }
            }
            if (!found) {
                appConfig.getList().add(bean);
            }

            appConfig.setUpdateTime(bean.getUpdateTime());
            String xml = JAXBUtils.dtoToXmlString(appConfig);
            CuratorSupport.updateNodeValue(client, AttributeSupport.APP_NODE, xml.getBytes());
        } catch (Exception e) {
            LOGGER.error("更新APP节点数据更新异常", e);
        }
    }

    /**
     * 同步MAIN_NODE节点配置文件
     */
    private void synchronizedMainConfig(String content) {
        try {
            File file = new File(AttributeSupport.syMainConfigFile);
            if (!file.exists()) {
                file.createNewFile();
                OutputStreamWriter write = new OutputStreamWriter(new FileOutputStream(file), "UTF-8");
                BufferedWriter writer = new BufferedWriter(write);
                writer.write(content);
                writer.close();
            } else {
                if (!needSynchronized(content.getBytes())) return;
                OutputStreamWriter write = new OutputStreamWriter(new FileOutputStream(file), "UTF-8");
                BufferedWriter writer = new BufferedWriter(write);
                writer.write(content);
                writer.close();
            }
        } catch (IOException e) {
            LOGGER.error("同步配置文件异常," + AttributeSupport.syMainConfigFile, e);
        }
    }

    /**
     * 同步APP_NODE节点配置文件
     */
    private void synchronizedAppConfig(String content) {
        try {
            File file = new File(AttributeSupport.syAppConfigFile);
            if (!file.exists()) {
                file.createNewFile();
            }
            OutputStreamWriter write = new OutputStreamWriter(new FileOutputStream(file), "UTF-8");
            BufferedWriter writer = new BufferedWriter(write);
            writer.write(content);
            writer.close();
        } catch (IOException e) {
            LOGGER.error("同步配置文件异常," + AttributeSupport.syAppConfigFile, e);
        }
    }

    private SynchronizedMainConfigBean getLocalSynchronizedConfig() {
        SynchronizedMainConfigBean bean = null;
        try {
            File file = new File(AttributeSupport.syMainConfigFile);
            String content = "";
            InputStreamReader read = new InputStreamReader(
                    new FileInputStream(file), "UTF-8");// 考虑到编码格式

            BufferedReader bufferedReader = new BufferedReader(read);
            String lineTxt = null;
            while ((lineTxt = bufferedReader.readLine()) != null) {
                content = content + lineTxt + "\n";
            }
            read.close();
            bean = JAXBUtils.xmlToDto(content, SynchronizedMainConfigBean.class);
        } catch (Exception e) {
            LOGGER.error("获得本地同步文件内容异常", e);
            bean = null;
        }
        return bean;
    }

    /**
     * 增加条件判断是否最新的配置信息
     */
    private boolean needSynchronized(byte[] data) {
        if (data == null) return false;
        try {
            SynchronizedMainConfigBean zkConfig = JAXBUtils.xmlToDto(new String(data), SynchronizedMainConfigBean.class);
            SynchronizedMainConfigBean localConfig = getLocalSynchronizedConfig();
            if (localConfig == null || localConfig.getUpdateTime() == null || localConfig.getUpdateTime().trim().equals("")
                    || TimeUtil.convertStr2Date(zkConfig.getUpdateTime()).after(TimeUtil.convertStr2Date(localConfig.getUpdateTime()))) {
                return true;
            }
        } catch (Exception e) {
            LOGGER.error("", e);
            return false;
        }
        return false;
    }

    private void initListener() throws Exception {
        ExecutorService pool = Executors.newFixedThreadPool(10);
        /**
         * 监听MAIN_NODE节点数据的变化情况
         */
        final NodeCache mainNodeCache = new NodeCache(client, AttributeSupport.MAIN_NODE, false);
        mainNodeCache.start(true);
        mainNodeCache.getListenable().addListener(
                new NodeCacheListener() {
                    public void nodeChanged() throws Exception {
                        synchronizedMainConfig(new String(mainNodeCache.getCurrentData().getData(), "UTF-8"));
                        watchMainNodeData(mainNodeCache);
                    }
                },
                pool
        );

        /**
         * 监听APP_NODE节点数据的变化情况
         */
        final NodeCache appNodeCache = new NodeCache(client, AttributeSupport.APP_NODE, false);
        appNodeCache.start(true);
        appNodeCache.getListenable().addListener(
                new NodeCacheListener() {
                    public void nodeChanged() throws Exception {
                        synchronizedAppConfig(new String(appNodeCache.getCurrentData().getData(), "UTF-8"));
                        watchAppNodeData(appNodeCache);
                    }
                },
                pool
        );

        /**
         * 监听MAIN_NODE的子节点的变化情况
         */
        final PathChildrenCache childrenCache = new PathChildrenCache(client, AttributeSupport.MAIN_NODE, true);
        childrenCache.start(PathChildrenCache.StartMode.POST_INITIALIZED_EVENT);
        childrenCache.getListenable().addListener(
                new PathChildrenCacheListener() {
                    public void childEvent(CuratorFramework client, PathChildrenCacheEvent event)
                            throws Exception {
//                        watchAppNodeState(client,event);
                        switch (event.getType()) {
                            case CHILD_ADDED:
//                                PushSupport.setInfo(new InfoBean(AttributeSupport.APP_NODE, new Date().toString(), EInfo.CREATE, AttributeSupport.APP_NODE + " CREATE"));
//                                startProcess(event);
                                break;
                            case CHILD_REMOVED:
//                                PushSupport.setInfo(new InfoBean(AttributeSupport.APP_NODE, TimeUtil.getUtcTime(), EInfo.DELETE, AttributeSupport.APP_NODE + " DELETE"));
//                                killProcess(event);
                                break;
//                            case CHILD_UPDATED:
//                                System.out.println("CHILD_UPDATED: " + event.getData().getPath());
//                                break;
                            default:
                                break;
                        }
                    }
                },
                pool
        );
    }

    public String[] getExec() {
        return exec;
    }

    public Process getChild() {
        return child;
    }

    public CuratorFramework getClient() {
        return client;
    }

    public void setChild(Process child) {
        this.child = child;
    }

    public AbstractLeaderSelectorClient getLeaderSelector() {
        return leaderSelector;
    }

    /**
     * APP_NODE节点增加后的处理
     */
    abstract public void startProcess(PathChildrenCacheEvent event);

    /**
     * APP_NODE节点删除后的处理
     */
    abstract public void killProcess(PathChildrenCacheEvent event);


    private CuratorListener getCuratorListener() {
        CuratorListener listener = new CuratorListener() {
            public void eventReceived(CuratorFramework curatorFramework, CuratorEvent curatorEvent) throws Exception {
                watchCurator(curatorFramework, curatorEvent);
            }
        };
        return listener;
    }

    private ConnectionStateListener getConnectionStateListener() {
        ConnectionStateListener listener = new ConnectionStateListener() {
            public void stateChanged(CuratorFramework curatorFramework, ConnectionState connectionState) {
                try {
                    watchConnectState(curatorFramework, connectionState);
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        };
        return listener;
    }

    public static void main(String[] args) {
        new MasterCurator(null);
    }

    abstract public void run();

    /**
     * 监听节点状态
     */
    abstract public void watchCurator(CuratorFramework client, CuratorEvent event);

    /**
     * 监听网络连接状态
     */
    abstract public void watchConnectState(CuratorFramework client, ConnectionState state) throws Exception;

    /**
     * 监听APP_NODE节点的变化情况
     */
//    abstract public void watchAppNodeState(CuratorFramework client, PathChildrenCacheEvent event);

    /**
     * 监听MAIN_NODE节点数据的变化情况
     */
    abstract public void watchMainNodeData(NodeCache nodeCache);

    /**
     * 监听APP_NODE节点数据的变化情况
     */
    abstract public void watchAppNodeData(NodeCache nodeCache);

    public ECuratorType getCuratorType(){
        return master? ECuratorType.MASTER:ECuratorType.CONCURRENT;
    }
}
